package com.zx.mes.hyl.controller;


import com.baomidou.mybatisplus.core.conditions.query.QueryWrapper;
import com.zx.mes.hyl.entity.CustomerAccount;
import com.zx.mes.hyl.producer.PayTransactionListener;
import com.zx.mes.hyl.response.ObjectRestResponse;
import com.zx.mes.hyl.rocket.mq.producer.SyncProducer;
import com.zx.mes.hyl.rocket.mq.producer.TransactionProducer;
import com.zx.mes.hyl.rocket.mq.yml.RocketMqProperties;
import com.zx.mes.hyl.service.CustomerAccountService;
import com.zx.mes.hyl.utils.FastJsonConvertUtil;
import lombok.extern.log4j.Log4j2;
import org.apache.rocketmq.client.consumer.listener.ConsumeOrderlyStatus;
import org.apache.rocketmq.client.producer.LocalTransactionState;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.client.producer.SendStatus;
import org.apache.rocketmq.client.producer.TransactionSendResult;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.remoting.common.RemotingHelper;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RequestParam;
import org.springframework.web.bind.annotation.RestController;

import java.io.UnsupportedEncodingException;
import java.math.BigDecimal;
import java.util.HashMap;
import java.util.Map;

/**
 * <p>
 * 前端控制器
 * </p>
 *
 * @author 华云龙
 * @since 2018-12-05
 */
@RestController
@RequestMapping("/customerAccount")
@Log4j2
public class CustomerAccountController extends BaseController<CustomerAccountService, CustomerAccount> {

    @Autowired
    private RocketMqProperties rocketMqProperties;

    private static final String PRODUCER_GROUP_NAME = "tx-long-pay-a-pay-producer-group-name";

    private static final String SYNC_PAYED_PRODUCER_GROUP_NAME = "common-long-pay-a-payed-producer-group-name";

    /**
     * userId,orderId,accountId,money
     *
     * @return ObjectRestResponse response
     */
    public ObjectRestResponse<CustomerAccount> pay(@RequestParam("userId") String userId,
                                                   @RequestParam("orderId") String orderId,
                                                   @RequestParam("accountId") String accountId,
                                                   @RequestParam("money") double money) {
        ObjectRestResponse.Builder<CustomerAccount> response = new ObjectRestResponse.Builder<>();
        /// 查看消费者的账户情况
        CustomerAccount customerAccount = this.biz.getOne(new QueryWrapper<CustomerAccount>().lambda().eq(CustomerAccount::getAccountId, accountId));
        BigDecimal payMoney = BigDecimal.valueOf(money);
        /// 检查余额
        if (customerAccount.getCurrentBalance().subtract(payMoney).doubleValue() > 0) {
            TransactionProducer transactionProducer = new TransactionProducer();
            try {
                /// 分布式事务开启
                PayTransactionListener listener = new PayTransactionListener();

                /// 配置生产者组
                transactionProducer.producerConfig(PRODUCER_GROUP_NAME, rocketMqProperties, listener);

                /// 发送消息
                Map<String, Object> params = new HashMap<>(8);
                params.put("accountId", customerAccount.getAccountId());
                params.put("version", customerAccount.getVersion());
                params.put("money", money);
                params.put("currentBalance", customerAccount.getCurrentBalance());
                params.put("userId", userId);
                params.put("orderId", orderId);
                RocketMqProperties.RocketMq rocketMq = rocketMqProperties.getMap().get(PRODUCER_GROUP_NAME);
                Message message = new Message(rocketMq.getTopic(), rocketMq.getTags(), rocketMq.getKeys(), FastJsonConvertUtil.convertObjectToJSONWithNullValue(params).getBytes(RemotingHelper.DEFAULT_CHARSET));
                TransactionSendResult sendResult = transactionProducer.sendMessage(message, params);
                if (sendResult.getSendStatus().equals(SendStatus.SEND_OK) && sendResult.getLocalTransactionState().equals(LocalTransactionState.COMMIT_MESSAGE)) {
                    /// 分布式事务发送成功
                    log.debug("本地事务与发送状态都OK!");
                    /// 发送普通同步信息,改变订单状 创建 --> 已支付
                    SyncProducer syncProducer = new SyncProducer();
                    syncProducer.producerConfig(SYNC_PAYED_PRODUCER_GROUP_NAME, rocketMqProperties);
                    RocketMqProperties.RocketMq mq = this.rocketMqProperties.getMap().get(SYNC_PAYED_PRODUCER_GROUP_NAME);
                    Map<String, Object> map = new HashMap<>(4);
                    map.put("status", "payed");
                    map.put("orderId", orderId);
                    Message mes = new Message(mq.getTopic(), mq.getTags(), mq.getKeys(), FastJsonConvertUtil.convertObjectToJSONWithNullValue(params).getBytes(RemotingHelper.DEFAULT_CHARSET));
                    SendResult result = syncProducer.sendMessage(mes);
                    if (!result.getSendStatus().equals(SendStatus.SEND_OK)) {
                        /// 消息发送不成功,人工做补偿
                        log.error("消息发送不成功,发送详细结果:[" + result + "],补偿信息:[" + params + "]");
                    }
                }
            } catch (UnsupportedEncodingException e) {
                e.printStackTrace();
                /// 余额不足
                log.error("编码异常,付款失败!");
                response.status(500)
                        .message("编码异常,付款失败!");
            } finally {
                /// 开启后要关闭
                transactionProducer.shutdown();
            }

        } else {
            /// 余额不足
            log.error("余额不足,付款失败!");
            response.status(500)
                    .message("余额不足,付款失败!");
        }

        return response.build();
    }
}

